package com.gitee.xhs.customWaterMark;

import com.gitee.xhs.*;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import static com.gitee.xhs.CustomDefaultTimestampAssignerSupplier.CUSTOM_DEFAULT_TIMESTAMP_ASSIGNER_SUPPLIER;
import static com.gitee.xhs.CustomDefaultWatermarkGeneratorSupplier.CUSTOM_DEFAULT_WATERMARK_GENERATOR_SUPPLIER;
import static com.gitee.xhs.WatermarkConstant.CUSTOM_WATERMARK_STRATEGY;

/**
 * @author BigHead
 * @date 2022/7/18 11:26
 * @description CustomTimestampAssignerSupplierMain
 */
public class CustomTimestampAssignerSupplierMain {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.webUi(new EnvironmentConfig(null));

        streamExecutionEnvironment
                .addSource(
                        WatermarkSourceFactory.defaultTransactionBoundOneSecondSource()
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .forGenerator(
                                        new CustomDefaultWatermarkGeneratorSupplier()
                                )
                                .withTimestampAssigner(
                                        new CustomDefaultTimestampAssignerSupplier()
                                )
                )
                .process(
                        new WatermarkRecordMakerProcessFunction(CUSTOM_WATERMARK_STRATEGY, CUSTOM_DEFAULT_WATERMARK_GENERATOR_SUPPLIER, CUSTOM_DEFAULT_TIMESTAMP_ASSIGNER_SUPPLIER)
                )
                .print();

        streamExecutionEnvironment.execute("CustomSerializableTimestampAssignerMain");
    }

}
